package com.sh.data.engine.domain.datadev.offline.runner;

import cn.hutool.extra.spring.SpringUtil;
import com.google.common.collect.Lists;
import com.sh.data.engine.domain.datadev.offline.model.enums.OfflineDevJobStatusEnum;
import com.sh.data.engine.domain.datadev.offline.model.enums.OfflineDevNodeTypeEnum;
import com.sh.data.engine.domain.datadev.offline.service.DatadevScriptService;
import com.sh.data.engine.domain.workspace.manager.model.domain.UserDomain;
import com.sh.data.engine.domain.workspace.manager.service.UserService;
import com.sh.data.engine.repository.dao.datadev.entity.DatadevJobEntity;
import com.sh.data.engine.repository.dao.datadev.entity.DatadevMenuNodeEntity;
import com.sh.data.engine.repository.dao.datadev.entity.DatadevScriptEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.exec.*;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang3.StringUtils;

import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;

/**
 * @author : fuchen
 * @date : 2022/3/7 16:43
 */
@Slf4j
public class OfflineDevShellJobRunner extends OfflineDevBaseJobRunner {

    private UserService userService = SpringUtil.getBean(UserService.class);
    private DefaultExecutor exec = new DefaultExecutor();
    private ExecuteWatchdog dog = null;
    private DatadevScriptEntity datadevScriptEntity;

    protected DatadevScriptService datadevScriptService =
        SpringUtil.getBean(DatadevScriptService.class);

    public OfflineDevShellJobRunner(
        DatadevJobEntity jobEntity,
        DatadevMenuNodeEntity menuNodeEntity,
        Map<Long, Integer> parentNodeIdAndLineStatusMap,
        List<Long> childNodeIdList,
        Map<Long, OfflineDevBaseJobRunner> allJobNodeMap,
        ExecutorService executorService) {
        super(
            jobEntity,
            menuNodeEntity,
            parentNodeIdAndLineStatusMap,
            childNodeIdList,
            allJobNodeMap,
            executorService);
    }

    public OfflineDevShellJobRunner(
        DatadevJobEntity jobEntity,
        DatadevMenuNodeEntity menuNodeEntity,
        Map<Long, Integer> parentNodeIdAndLineStatusMap,
        List<Long> childNodeIdList,
        Map<Long, OfflineDevBaseJobRunner> allJobNodeMap,
        ExecutorService executorService,
        DatadevScriptEntity datadevScriptEntity) {
        super(
            jobEntity,
            menuNodeEntity,
            parentNodeIdAndLineStatusMap,
            childNodeIdList,
            allJobNodeMap,
            executorService);
        this.datadevScriptEntity = datadevScriptEntity;
    }

    @Override
    public void cancel() {
        // todo

    }

    @Override
    public int runJob() throws Exception {
        String userId = jobEntity.getCreatorId();
        String s = String.format("节点:[%s]开始执行", menuNodeEntity.getName());
        s = formatLogWithTime(s);
        this.writeLog(Lists.newArrayList(s));

        String finishedLog = String.format("节点:[%s]结束执行", menuNodeEntity.getName());
        if (datadevScriptEntity == null) {
            datadevScriptEntity =
                datadevScriptService.getDatadevScriptByNodeId(menuNodeEntity.getId());
        }
        datadevScriptEntity =
            datadevScriptService.getDatadevScriptByNodeId(menuNodeEntity.getId());
        if (null == datadevScriptEntity) {
            String log2 = formatLogWithTime("该节点无配置");

            finishedLog = formatLogWithTime(finishedLog);

            this.writeLog(Lists.newArrayList(log2, finishedLog));

            return OfflineDevJobStatusEnum.SUCCESS.getCode();
        }
        String script = datadevScriptEntity.getScript();
        UserDomain userDomain = userService.getUserById(userId);
        String user = userDomain.getCmdUser();
        String userTmpDirPath =
            String.format(
                "/tmp/%s/%s",
                user, OfflineDevNodeTypeEnum.of(menuNodeEntity.getNodeType()).getNodeDesc());
        File dirFile = new File(userTmpDirPath);
        if (!dirFile.exists()) {
            dirFile.mkdirs();
            CommandLine commandLine = new CommandLine("chown");
            commandLine.addArgument("-R");
            commandLine.addArgument(String.format("%s:%s", user, user));
            commandLine.addArgument(String.format("/tmp/%s", user));
            try {
                this.exec.execute(commandLine);
            } catch (IOException e) {
                String logExec = String.format("创建用户目录:[%s]失败", userTmpDirPath);
                logExec = formatLogWithTime(logExec);
                this.writeLog(Lists.newArrayList(logExec));
            }
        }
        if (StringUtils.isNotBlank(script)) {
            Integer nodeType = menuNodeEntity.getNodeType();
            script = script.replaceAll("\r\n", "\n").replaceAll("\r", "\n");
            String fileSuffix = ".sh";
            String command = "/bin/bash";
            OfflineDevNodeTypeEnum nodeTypeEnum = OfflineDevNodeTypeEnum.of(nodeType);
            if (nodeTypeEnum.getNodeDesc().equals(OfflineDevNodeTypeEnum.PYTHON2.getNodeDesc())) {
                fileSuffix = ".py";
                command = "/usr/bin/python2";
            } else if (nodeTypeEnum.getNodeDesc().equals(OfflineDevNodeTypeEnum.PYTHON3.getNodeDesc())) {
                fileSuffix = ".py";
                command = "/usr/bin/python3";
            } else if (nodeTypeEnum.getNodeDesc().equals(OfflineDevNodeTypeEnum.PYSPARK.getNodeDesc())) {
                fileSuffix = ".py";
                command = "/usr/bin/spark-submit";
            }

            String shFilePath = userTmpDirPath + File.separator + jobEntity.getId() + fileSuffix;
            try {
                FileUtils.write(new File(shFilePath), script, "UTF-8");
                CommandLine commandLine = new CommandLine("chown");
                commandLine.addArgument(String.format("%s:%s", user, user));
                commandLine.addArgument(shFilePath);
                exec.execute(commandLine);
            } catch (IOException e) {
                log.error(e.getMessage(), e);

                String s1 = String.format("生成脚本文件异常，任务结束");
                s1 = formatLogWithTime(s1);
                this.writeLog(Lists.newArrayList(s1));
            }

            dog = new ExecuteWatchdog(60 * 60 * 1000);
            CommandLine commandline = new CommandLine("su");
            commandline.addArgument(user);
            commandline.addArgument("-s");
            commandline.addArgument(command);
            commandline.addArgument(shFilePath);
            exec.setStreamHandler(new PumpStreamHandler(this.createLogOutStream()));
            exec.setWatchdog(dog);
            this.writeLog((Lists.newArrayList(formatLogWithTime("即将执行命令：" + commandline))));
            int exitValue = 0;
            try {
                exitValue = exec.execute(commandline);
            } catch (Exception e) {
                this.writeLog(Lists.newArrayList("\n", "命令执行异常", e.getMessage()));
                FileUtils.deleteQuietly(new File(shFilePath));
                return OfflineDevJobStatusEnum.SUCCESS.getCode();
            }
            this.writeLog(Lists.newArrayList("\n", "命令执行结束，退出码：" + exitValue));
            return OfflineDevJobStatusEnum.SUCCESS.getCode();
        }
        return OfflineDevJobStatusEnum.SUCCESS.getCode();
    }

    private LogOutputStream createLogOutStream() {

        return new LogOutputStream() {
            @Override
            protected void processLine(String line, int logLevel) {
                writeLog(Lists.newArrayList(line));
            }
        };
    }
}
